package com.example.sparksqltest.demos.service;


import com.example.sparksqltest.demos.dao.MovieRating;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.springframework.stereotype.Service;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.util.List;
import java.util.stream.Collectors;

@Service
public class MovieDataService {

    private static final Logger logger = LoggerFactory.getLogger(MovieDataService.class);

    public List<MovieRating> getMovieRatings(SparkSession sparkSession) {
        logger.info("正在加载电影数据...");
        Dataset<Row> movieData = loadMovieData(sparkSession);
        logger.info("电影数据加载完成，行数: {}", movieData.count());
        logger.info("正在转换电影数据为MovieRating对象...");
        List<MovieRating> ratings = transformToMovieRating(movieData);
        logger.info("转换完成，共生成{}个MovieRating对象。", ratings.size());
        return ratings;
    }

    private Dataset<Row> loadMovieData(SparkSession sparkSession) {
        logger.info("正在从JDBC源读取数据...");
        return sparkSession.read()
                .format("jdbc")
                .option("url", "jdbc:mysql://localhost:3306/bigdata?useSSL=false&serverTimezone=UTC")
                .option("dbtable", "movie_ratings")
                .option("user", "root")
                .option("password", "woai782013")
                .option("driver", "com.mysql.cj.jdbc.Driver")
                .load();
    }

    private List<MovieRating> transformToMovieRating(Dataset<Row> movieData) {
        return movieData.collectAsList().stream()
                .map(row -> {
                    MovieRating rating = new MovieRating(
                            row.getAs("genre"),
                            row.getAs("Gender"),
                            new BigDecimal(row.getAs("rating").toString())
                    );
                    logger.debug("已转换行数据为MovieRating对象: {}", rating);
                    return rating;
                })
                .collect(Collectors.toList());
    }
}
